// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: jing

#include "coin2/exchange/coinone/feed_v2/processor.h"

#include <string>
#include <utility>

#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/coinone/feed_v2/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::coinone::feed_v2::impl {

void CoinoneFeedProcessor::OnTopicRecord(const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    THROW() << "JSON parse error: " << GetParseError_En(document.GetParseError());
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (!document.HasMember("event") && document.HasMember("channel")) {
    const std::string& channel = document["channel"].GetString();
    if (!CheckChannelFilter(channel)) {
      return;
    }

    const std::string& native_symbol = document["trading_pair"].GetString();
    if (channel == "trades") {
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        return;
      }
      Trade* trade = upd_mgr_.GetTradeMutable();
      CoinoneFeedUpdateParser::ParseTrade(document, trade);
      upd_mgr_.PublishTrade(native_symbol, false);
    } else if (channel == "orderbook") {
      std::string native_symbol_upper = native_symbol;
      std::transform(native_symbol_upper.begin(), native_symbol_upper.end(), native_symbol_upper.begin(), ::toupper);
      auto* book = upd_mgr_.GetBookMutable(native_symbol_upper);
      if (book != nullptr) {
        CoinoneFeedUpdateParser::ParseBook(book, document);
        upd_mgr_.PublishBook();
      }
    } else {
      LOG(INFO) << "[Coinone] unknown msg for " << packet.topic_name << ": "
                << packet.payload.data();
    }
  } else {
    // meta msg like subscription response...
    LOG(INFO) << "[Coinone Feed Processor] meta msg for " << packet.topic_name << ": "
              << packet.payload.data();
  }

  upd_mgr_.onEndPacket(packet);
}  // namespace coin2::exchange::coinone::feed_v2::impl
}  // namespace coin2::exchange::coinone::feed_v2::impl
